// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/connectors/hive/storage_adapters/s3fs/S3Config.h>

#include <pollux/common/config/config.h>
#include <pollux/connectors/hive/storage_adapters/s3fs/S3Util.h>

namespace kumo::pollux::filesystems {

std::string S3Config::cacheKey(
    std::string_view bucket,
    std::shared_ptr<const config::ConfigBase> config) {
  auto bucketEndpoint = bucketConfigKey(Keys::kEndpoint, bucket);
  if (config->valueExists(bucketEndpoint)) {
    return fmt::format(
        "{}-{}", config->get<std::string>(bucketEndpoint).value(), bucket);
  }
  auto baseEndpoint = baseConfigKey(Keys::kEndpoint);
  if (config->valueExists(baseEndpoint)) {
    return fmt::format(
        "{}-{}", config->get<std::string>(baseEndpoint).value(), bucket);
  }
  return std::string(bucket);
}

S3Config::S3Config(
    std::string_view bucket,
    const std::shared_ptr<const config::ConfigBase> properties)
    : bucket_(bucket) {
  for (int key = static_cast<int>(Keys::kBegin);
       key < static_cast<int>(Keys::kEnd);
       key++) {
    auto s3Key = static_cast<Keys>(key);
    auto value = S3Config::configTraits().find(s3Key)->second;
    auto configSuffix = value.first;
    auto configDefault = value.second;

    // Set bucket S3 config "hive.s3.bucket.*" if present.
    std::stringstream bucketConfig;
    bucketConfig << kS3BucketPrefix << bucket << "." << configSuffix;
    auto configVal = static_cast<std::optional<std::string>>(
        properties->get<std::string>(bucketConfig.str()));
    if (configVal.has_value()) {
      config_[s3Key] = configVal.value();
    } else {
      // Set base config "hive.s3.*" if present.
      std::stringstream baseConfig;
      baseConfig << kS3Prefix << configSuffix;
      configVal = static_cast<std::optional<std::string>>(
          properties->get<std::string>(baseConfig.str()));
      if (configVal.has_value()) {
        config_[s3Key] = configVal.value();
      } else {
        // Set the default value.
        config_[s3Key] = configDefault;
      }
    }
  }
  payloadSigningPolicy_ =
      properties->get<std::string>(kS3PayloadSigningPolicy, "Never");
}

std::optional<std::string> S3Config::endpointRegion() const {
  auto region = config_.find(Keys::kEndpointRegion)->second;
  if (!region.has_value()) {
    // If region is not set, try inferring from the endpoint value for AWS
    // endpoints.
    auto endpointValue = endpoint();
    if (endpointValue.has_value()) {
      region = parseAWSStandardRegionName(endpointValue.value());
    }
  }
  return region;
}

} // namespace kumo::pollux::filesystems
